package com.zetyun.tiger.connector.cdc;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.zetyun.rt.sdk.model.RtEvent;
import lombok.AllArgsConstructor;
import lombok.SneakyThrows;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

import java.util.List;
import java.util.stream.Collectors;

@AllArgsConstructor
public class FormatRedisMapper implements RedisMapper<RtEvent> {
    private RedisCommand redisCommand;

    private String additionalKey;

    private List<String> keyNames;

    private ObjectMapper om;

    @Override
    public RedisCommandDescription getCommandDescription() {

        return new RedisCommandDescription(redisCommand, additionalKey);
    }

    @Override
    public String getKeyFromData(RtEvent data) {
        return keyNames.stream()
                .map(k -> data.getField(k, String.class))
                .collect(Collectors.joining("_"));
    }

    @SneakyThrows
    @Override
    public String getValueFromData(RtEvent data) {

        return om.writeValueAsString(data.getFields());

    }
}
